package xubo.wangcaifeng.love.method

import ch.hsr.geohash.GeoHash
import org.apache.commons.lang.StringUtils
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SQLContext
import xubo.wangcaifeng.love.Utils.{JedisConnectionPool, JedisPool, TU}

object Need3Success {
  def main(args: Array[String]): Unit = {
    if (args.length != 4) {
      println(
        """
          |cn.dmp.report.AnalyseProvince
          |params:
          | dataInputPath parquet输入路径
          | dictInputPath   appid匹配规则
          | stopwordsInputPath  不需要词汇的路径
          | outputPath  结果输出路径
        """.stripMargin)
      sys.exit()
    }
    val Array(dataInputPath, outputPath, dictInputPath, stopwordsInputPath) = args
    val conf = new SparkConf()
      .setAppName("按照查询")
      .setMaster("local[*]")
      .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    val sc = new SparkContext(conf)
    val sqlc = new SQLContext(sc)
    // 加载本地客户端的配置信息 core-default.xml hdfs-default.xml
    val hadoopConfiguration = sc.hadoopConfiguration
    val fs = FileSystem.get(hadoopConfiguration) // fs 就是一个本地的文件操作系统
    val destDir = new Path(outputPath)
    if (fs.exists(destDir)) { // 如果存在，删之
      fs.delete(destDir, true) // 递归删除
    }
    //读取标准的appid匹配规则
    val dictMap = sc.textFile(dictInputPath).map(_.split("\t", -1)).filter(_.length >= 5).map(arr => (arr(4), arr(1))).collect().toMap
    val dictBC = sc.broadcast(dictMap)
    //停用字典文件,把它广播出去
    val stopWordMap = sc.textFile(stopwordsInputPath).map(line => (line, 0)).collect().toMap
    val stopWordBC = sc.broadcast(stopWordMap)
    //读取文件
    val frame = sqlc.read.parquet(dataInputPath)
    val value = frame.where(TU.hasUserIdCondition).mapPartitions(iter => {
      val jedis = JedisConnectionPool.getConnection()
      val result = iter.map({
        row =>
          //标签逻辑
          //定义一个map将所有得到的标签都放入map中
          var tmap = Map[String, Int]()
          //广告位类型
          val adType = row.getAs[Int]("adspacetype")
          val adTypeName = row.getAs[String]("adspacetypename")
          if (adType >= 10) tmap += "LC" + adType -> 1 else tmap += "LC0" + adType -> 1
          if (StringUtils.isNotEmpty(adTypeName)) tmap += "LN" + adTypeName -> 1
          //渠道
          val chnId = row.getAs[Int]("adplatformproviderid")
          if (chnId > 0) tmap += "CN" + chnId -> 1
          //app
          val appId = row.getAs[String]("appid")
          val appName = row.getAs[String]("appname")
          val newAppName = if (StringUtils.isEmpty(appName)) {
            if (StringUtils.isNotEmpty(appId)) {
              dictBC.value.getOrElse(appId, appId)
            } else ""
          } else {
            appName
          }
          if (StringUtils.isNotEmpty(newAppName)) tmap += "App" + newAppName -> 1
          //设备
          val os = row.getAs[Int]("client")
          val ntwkName = row.getAs[String]("networkmannername")
          val isp = row.getAs[String]("ispname")

          os match {
            case 1 => tmap += "D00010001" -> 1
            case 2 => tmap += "D00010002" -> 1
            case 3 => tmap += "D00010003" -> 1
            case _ => tmap += "D00010004" -> 1
          }
          ntwkName.toUpperCase() match {
            case "WIFI" => tmap += "D00020001" -> 1
            case "4G" => tmap += "D00020002" -> 1
            case "3G" => tmap += "D00020003" -> 1
            case "2G" => tmap += "D00020004" -> 1
            case _ => tmap += "D00020005" -> 1
          }
          isp match {
            case "移动" => tmap += "D00030001" -> 1
            case "联通" => tmap += "D00030002" -> 1
            case "电信" => tmap += "D00030003" -> 1
            case _ => tmap += "D00040004" -> 1
          }
          //关键字
          val keyWords = row.getAs[String]("keywords")
          keyWords.split("|").filter(key => key.length >= 3 && key.length <= 8 && !stopWordBC.value.contains(key)).foreach(key => tmap += "K" + key -> 1)
          //地域
          val pName = row.getAs[String]("provincename")
          val cName = row.getAs[String]("cityname")
          if (StringUtils.isNotEmpty(pName)) tmap += "ZP" + pName -> 1
          if (StringUtils.isNotEmpty(cName)) tmap += "ZC" + cName -> 1

          //商圈标签
          val long = row.getAs[String]("long")
          val lat = row.getAs[String]("lat")


          if (long.nonEmpty && long >= "73.66" && long <= "135.05" && lat >= "3.86" && lat <= "53.55" && lat.nonEmpty) {

            val geoHash = GeoHash.withCharacterPrecision(long.toDouble, lat.toDouble, 8).toBase32
            if (StringUtils.isNotEmpty(geoHash.toString)) {
              val local = jedis.get(geoHash.toString)
              tmap += local -> 1
            }

          }

          // (1, List((APP爱奇艺,1),(ZP湖北省,1),(CN,1001)))
          // (1, List((D00010001,1),(ZP湖北省,1)))
          //返回一个（UID，Map[String,Int])
          (TU.getUserId(row), tmap.toList)
      })
      jedis.close()
      result
    })


    /**
      * value (1, List((APP爱奇艺,1),(ZP湖北省,1),(CN,1001)))
      * (1, List((D00010001,1),(ZP湖北省,1)))
      */
    value.reduceByKey((list1, list2) => (list1 ++ list2)
      //value 的 值是   (1，List((APP爱奇艺,1),(ZP湖北省,1),(CN,1001), (D00010001,1),(ZP湖北省,1)//前面是第一条的数据,后面是第二条的(xx,xx,xx,xx,xx)))
      /**
        * gorupBy之后的结果   这个是map
        * // Map((APP爱奇艺,List((APP爱奇艺, 1))),(ZP湖北省,List((ZP湖北省,2), (ZP湖北省,1))),(D00010001,List((D00010001, 1)))))
        */
      .groupBy(_._1)
      //对groupBy之后的结果进行求和
      .mapValues(_.map((_._2)).sum).toList
    ).coalesce(1).saveAsTextFile(outputPath)
    sc.stop()

  }

}
